package com.lecosa.flink;

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object ReduceTest {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    import org.apache.flink.streaming.api.scala._
    val lines: DataStream[String] = env.socketTextStream("tfmaster01",9999)
    lines.flatMap(_.split(" "))
      .map((_,1))
      .keyBy(0)
        .reduce(new ReduceFunction[(String, Int)] {
          override def reduce(tp1: (String, Int), tp2: (String, Int)): (String, Int) = {
            (tp1._1,tp1._2+tp2._2)
          }
        }).print()
//      .reduce((tp1,tp2)=>{
//        (tp1._1,tp1._2+tp2._2)
//      }).print()
    env.execute()
  }

}
